package com.compus.mqtt;

import cn.hutool.json.JSON;
import cn.hutool.json.JSONUtil;
import com.compus.entity.dto.CarPositionDto;
import com.compus.entity.po.CarPosition;
import com.compus.service.CarPositionService;
import com.compus.service.OrderService;
import com.compus.utils.GCJ02_WGS84Utils;
import com.compus.utils.LocateInfo;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@ConditionalOnProperty(name = "campus.enabled", havingValue = "true")
public class MqttApplication implements InitializingBean, DisposableBean {

    @Resource
    private CarPositionService carPositionService;
    @Resource
    OrderService orderService;

    @Value("${campus.broker}")
    private String broker;
    // TLS/SSL
    // String broker = "ssl://broker.emqx.io:8883";
    @Value("${campus.username}")
    private String username;
    @Value("${campus.password}")
    private String password;
    String clientid = "location_client";

    @Value("${campus.topic-accept}")
    private String locationTopic;
    @Value("${campus.topic-order-accept}")
    private String orderTopic;

    MqttClient client = null;
    private static final ExecutorService executorService = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(50000));

    @Override
    public void destroy() throws Exception {
        try {
            if (client != null && client.isConnected()) {
                client.disconnect();
                client.close();
            }
            boolean awaitedTermination = executorService.awaitTermination(10, TimeUnit.SECONDS);
            if (!awaitedTermination) {
                executorService.shutdownNow();
            }
        } catch (Exception e) {
            log.error("mqtt close error", e);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        client = new MqttClient(broker, clientid, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
//        client.connect(options);
        int qos = 0;
        try {
            // 连接参数
            options.setConnectionTimeout(60);
            options.setKeepAliveInterval(60);
            // 设置回调
            client.connect(options);
            client.subscribe(locationTopic, qos, (topic, message) -> {
                log.info("topic: {}", topic);
                String msg = new String(message.getPayload());
                executorService.execute(() -> processMessage(msg));
                log.info("topic: {},location message: {}",topic, msg);
            });
            client.subscribe(orderTopic, qos, (topic, message) -> {
                String msg = new String(message.getPayload());
                log.info("topic: {},order message: {}",topic, msg);
                executorService.execute(() -> orderService.processOrder(msg));
            });
        } catch (Exception e) {
            log.error("mqtt connect error", e);
        }
    }

    public static final DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private void processMessage(String msg) {
        try {
            CarPositionDto dto = JSONUtil.toBean(msg, CarPositionDto.class);
            CarPosition position = new CarPosition();
            position.setCarId(Long.valueOf(dto.getCarId()));
            String[] split = dto.getPosition().split(",");
            LocateInfo locateInfo = GCJ02_WGS84Utils.wgs84_To_Gcj02(Double.parseDouble(split[1]), Double.parseDouble(split[0]));
            //经度,维度
            position.setPosition(locateInfo.getLongitude() + "," + locateInfo.getLatitude());
            position.setCreateTime(LocalDateTime.parse(dto.getTime(), timeFormatter));
            carPositionService.save(position);
            log.info("processMessage: {}", msg);
        } catch (Exception e) {
            log.error("processMessage error", e);
        }
    }


}
